package com.zh.freechat.domain.message.repository;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.zh.freechat.domain.chat.model.ChatMessage;
import com.zh.mycommon.value.Tuple2;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.SerializationUtils;
import org.jetbrains.annotations.Nullable;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * 消息发送记录表：作用是防重表+发送消息留踪迹。
 * 离线消息表：只暂存离线的消息。
 * <p>
 * 当client发送消息时：
 * 1、当server接收到消息时，会先持久化，再进行消息投递和推送等
 * 2、若peer在线：推送消息给 peer client端
 * 3、若 peer离线：则消息会存入离线表。
 * 当peer上线后，server会拉取离线表中的消息，再重新推送消息给 peer client端
 * <p>
 * 当server推送消息给peer端时：
 * 1、会触发一个Status_SENT事件消息，然后会记录下这次“发送消息记录”
 * 2、会移除离线消息表中的这条消息（假如存在的话）
 * 3、更新chat_msg表的这个message状态字段
 *
 * @author Zhou
 * @date 9:15 2022/7/23
 */
@Slf4j
@Repository
public class ChatMsgSendRecordRepository {
    static final String ch_msg_send_record = "ch_msg_send_record";
    static final String Sql_exist_sent = "select count(*) from " + ch_msg_send_record + " where msg_id=? ";
    static final String Sql_insert_sent = "INSERT INTO " + ch_msg_send_record + " (msg_id,receiver,success, create_time) VALUES (?, ?, ?, ?)";

    static final String ch_msg_offline_record = "ch_msg_offline_record";
    static final String Sql_insert_offline_msg = "INSERT INTO " + ch_msg_offline_record + " (msg_id,receiver,msg_bytes, create_time) VALUES (?,?,?,?)";
    static final String Sql_delete_offline_msg = "delete from " + ch_msg_offline_record + " where msg_id=? AND receiver=? ";
    static final String Sql_select_offline_msg = "select * from " + ch_msg_offline_record + " where msg_id=? AND receiver=? ";
    static final String Sql_select_offline_msg_by_user = "select * from " + ch_msg_offline_record + " where receiver=? AND create_time>=? order by create_time asc limit ?";

    private final JdbcTemplate jdbcTemplate;

    public ChatMsgSendRecordRepository(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }


    /**
     * msg的发送记录是否已存在
     *
     * @param msgId
     * @return
     */
    public boolean existSentByMsgId(String msgId) {
        Integer count = jdbcTemplate.queryForObject(Sql_exist_sent, Integer.class, msgId);
        return count > 0;
    }

    /**
     * 保存msg的发送记录
     *
     * @param msgId
     * @return
     */
    public boolean insertSentRecord(long msgId, String receiver, boolean success) {
        try {
            jdbcTemplate.update(Sql_insert_sent, msgId, receiver, success, LocalDateTime.now());
            return true;
        } catch (Exception ex) {
            log.error("save SentRecord 失败 msgId={}，err={}", msgId, ex.getMessage());
            return false;
        }
    }

    /**
     * 加入到离线消息表中
     * 此操作是幂等的
     *
     * @param message    message
     * @return bool 是否加入成功
     */
    public boolean insertOfflineRecord(ChatMessage message) {
        long msgId = message.getId();
        String userId = message.getReceiver();
        try {
            byte[] bytes = SerializationUtils.serialize(message);
            jdbcTemplate.update(Sql_insert_offline_msg, msgId, userId, bytes, LocalDateTime.now());
            return true;
        } catch (Exception ex) {
            log.error("save OfflineRecord 失败 msgId={}，err={}", msgId, ex.getMessage());
            return false;
        }
    }

    /**
     * 批量插入离线消息记录（幂等）
     * @param messages 消息列表
     * @return 插入成功的条数
     */
    public int[] batchInsertOfflineRecord(List<ChatMessage> messages) {
        if (CollectionUtils.isEmpty(messages)) {
            return new int[0];
        }

        // 组装批量参数（每个元素对应一条记录的参数）
        final LocalDateTime now = LocalDateTime.now();
        List<Object[]> batchParams = messages.stream()
                .map(msg -> new Object[]{
                        msg.getId(),
                        msg.getReceiver(),
                        SerializationUtils.serialize(msg),
                        now
                })
                .collect(Collectors.toList());

        // 执行批量更新（一行代码完成）
        try {
            return jdbcTemplate.batchUpdate(Sql_insert_offline_msg, batchParams);
        } catch (Exception ex) {
            log.error("批量插入失败，条数:{}，错误:{}", messages.size(), ex.getMessage(), ex);
            throw new RuntimeException("批量插入离线消息异常", ex);
        }
    }

    public void deleteOfflineRecord(long msgId, String userId) {
        try {
            jdbcTemplate.update(Sql_delete_offline_msg, msgId, userId);
        } catch (Exception ex) {
            log.error("delete OfflineRecord 失败 msgId={}，err={}", msgId, ex.getMessage());
        }
    }

    @Nullable
    public Tuple2<ChatMessage, LocalDateTime> selectOfflineRecord(long msgId, String userId) {
        try {
            MsgOfflineRecordPO recordPO = jdbcTemplate.queryForObject(Sql_select_offline_msg, new BeanPropertyRowMapper<>(MsgOfflineRecordPO.class), msgId, userId);
            if (recordPO == null || recordPO.msg_bytes.length == 0) {
                return null;
            }
            return new Tuple2<>(SerializationUtils.deserialize(recordPO.msg_bytes), recordPO.create_time);
        } catch (Exception ex) {
            log.error("select OfflineRecord 失败 msgId={}，err={}", msgId, ex.getMessage());
            return null;
        }
    }

    public List<Tuple2<ChatMessage, LocalDateTime>> selectOfflineRecordsByUserId(String userId, LocalDateTime startTime, int limit) {
        try {
            List<MsgOfflineRecordPO> recordList = jdbcTemplate.query(Sql_select_offline_msg_by_user, new BeanPropertyRowMapper<>(MsgOfflineRecordPO.class), userId, startTime, limit);
            return recordList.stream()
                    .filter(ele -> ele.msg_bytes.length > 0)
                    .map(ele -> new Tuple2<>(SerializationUtils.<ChatMessage>deserialize(ele.msg_bytes), ele.create_time))
                    .collect(Collectors.toList());
        } catch (Exception ex) {
            log.error("select OfflineRecord by user 失败 userId={}，err={}", userId, ex.getMessage());
            return Collections.emptyList();
        }
    }


    @Data
    @TableName(value = ch_msg_send_record) //range分区
    static class MsgSendRecordPO implements Serializable {
        //复合主键(msg_id,receiver)
        @TableId(value = "msg_id", type = IdType.INPUT)
        public String msg_id;
        @TableId(value = "receiver", type = IdType.INPUT)
        public String receiver;

        public boolean success;

        public LocalDateTime create_time;
    }
    @Data
    @TableName(value = ch_msg_offline_record)
    static class MsgOfflineRecordPO implements Serializable {
        //ID,自增
        @TableId(type = IdType.AUTO)
        public Long id;

        public String msg_id;

        public String receiver;

        public byte[] msg_bytes;

        public LocalDateTime create_time;
    }
}


